package cn.chenchendemospringboot.stock_demo.service.Impl;

import cn.chenchendemospringboot.stock_demo.entity.Inventory;
import cn.chenchendemospringboot.stock_demo.entity.KafkaEventProcess;
import cn.chenchendemospringboot.stock_demo.entity.Order;
import cn.chenchendemospringboot.stock_demo.mapper.InventoryMapper;
import cn.chenchendemospringboot.stock_demo.mapper.KafkaEventProcessMapper;
import cn.chenchendemospringboot.stock_demo.service.KafkaEventProcessService;
import com.baomidou.mybatisplus.core.conditions.query.QueryWrapper;
import com.google.gson.Gson;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Service;
import org.springframework.transaction.annotation.Transactional;
import org.springframework.util.CollectionUtils;

import java.util.List;
import java.util.stream.Collectors;

@Service
public class KafkaEvenProcessServiceImpl implements KafkaEventProcessService {
    @Autowired
    KafkaEventProcessMapper kafkaEventProcessMapper;

    @Autowired
    KafkaEventProcessImplMybatisPlus kafkaEventProcessImplMybatisPlus;

    @Autowired
    InventoryService inventoryService;

    @Override
    public void save(KafkaEventProcess kafkaEventProcess) {
        kafkaEventProcessMapper.insert(kafkaEventProcess);
    }

    @Override
    public List<KafkaEventProcess> selectByStatus(String status) {
        QueryWrapper<KafkaEventProcess> queryWrapper = new QueryWrapper<>();
        queryWrapper.eq("status", status);
        return kafkaEventProcessMapper.selectList(queryWrapper);
    }


    /**
     * 定时任务轮询
     * 查询消息表New
     * 执行业务逻辑 批量保存
     * 更新消息为Process
     */
    @Transactional
    public void process() {
        List<KafkaEventProcess> kafkaEventProcessList = selectByStatus("NEW");
        Gson gson = new Gson();
        if(!CollectionUtils.isEmpty(kafkaEventProcessList)){
            List<Inventory> inventoryList = kafkaEventProcessList.stream().map(event -> {
                // 将 payload 字段转换为 Inventory 对象
                Order order = gson.fromJson(event.getPayload(), Order.class);
                Inventory inventory = new Inventory();
                inventory.setProductName(String.valueOf(order.getOrderId()));
                inventory.setKafkaEventProcessId(event.getId());

                inventory.setStockQuantity(4);
                return inventory;
            }).collect(Collectors.toList());
            boolean b = inventoryService.saveBatch(inventoryList);

            if (b) {
                // 如果批量保存成功，更新 KafkaEventProcess 的 status
                List<KafkaEventProcess> updatedKafkaEvents = inventoryList.stream().map(inventory -> {
                    KafkaEventProcess updatedEvent = new KafkaEventProcess();
                    updatedEvent.setId(inventory.getKafkaEventProcessId());
                    updatedEvent.setStatus("Process"); // 设置新的状态
                    return updatedEvent;
                }).collect(Collectors.toList());

                // 更新 KafkaEventProcess 的状态
                if(!CollectionUtils.isEmpty(updatedKafkaEvents)){
                    kafkaEventProcessImplMybatisPlus.updateBatchById(updatedKafkaEvents);
                }

            }
        }


        //        for(KafkaEventProcess kafkaEventProcess:kafkaEventProcessList ){
//            //取到消息体的数据
//            Order order = gson.fromJson(kafkaEventProcess.getPayload(), Order.class);
//            //库存服务的业务逻辑
//            Inventory inventory = new Inventory();
//            inventory.setProductName(String.valueOf(order.getOrderId()));
//            inventory.setStockQuantity(4);
//            inventoryService.saveInventory(inventory);
//            kafkaEventProcessMapper.updateById(kafkaEventProcess,)
//        }
    }
}
